package com.storm.lesson.lesson;

import java.util.Map;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;

public class MyBolt implements IRichBolt {
	private static final long serialVersionUID = 1L;

	// 使用OutputCollector类的emit()方法指定发射的流
	OutputCollector collector = null;

	int num = 0;
	String valueString = null;

	@Override
	public void cleanup() {
		// TODO Auto-generated method stub

	}

	/**
	 * Bolt的主要方法是execute()方法，该方法将一个新的元组作为输入。Bolt使用OutputCollector对象发射新的元组.
	 */
	@Override
	public void execute(Tuple input) {

		try {
			valueString = input.getStringByField("log");

			if (valueString != null) {
				num++;
				System.out.println(Thread.currentThread().getName()
						+ "     lines: " + num + "   session_id: "
						+ valueString.split("\t")[1]);
			}
			// Spout必须为它们处理的每个元组调用OutputCollector类的ack()方法，以便storm知道什么时候元组会完成
			collector.ack(input);
		} catch (Exception e) {
			collector.fail(input);
			e.printStackTrace();
		}
	}

	@Override
	public void prepare(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		this.collector = collector;
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields(""));
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		// TODO Auto-generated method stub
		return null;
	}

}
